package org.budo.warehouse.logic.consumer.jdbc;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.annotation.Resource;

import org.budo.graph.annotation.SpringGraph;
import org.budo.support.java.concurrent.executor.ExecutorUtil;
import org.budo.support.spring.util.SpringUtil;
import org.budo.warehouse.logic.api.DataMessage;
import org.budo.warehouse.logic.bean.LogicDynamicBeanProvider;
import org.budo.warehouse.service.api.JdbcExecuteService;
import org.budo.warehouse.service.entity.Pipeline;
import org.springframework.aop.framework.AopContext;

import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;

/**
 * @author limingwei
 */
@ToString
@Slf4j
@Getter
@Setter
public class JdbcDataConsumer extends AbstractJdbcDataConsumer {
    @Resource
    private LogicDynamicBeanProvider logicDynamicBeanProvider;

    /**
     * 为空时主线程执行
     */
    private ExecutorService executorService;

    @SpringGraph
    @Override
    public void consume(DataMessage dataMessage) {
        List<SqlUnit> sqlUnits = this.buildSql(dataMessage);

        if (null == sqlUnits || sqlUnits.isEmpty()) { // 空的
            log.debug("#47 sqlUnits={}, dataMessage={}, this={}", sqlUnits, dataMessage, this);

            return;
        }

        JdbcDataConsumer _this = (JdbcDataConsumer) AopContext.currentProxy();

        // 按批次大小分方法，便于性能统计优化
        int len = sqlUnits.size();
        if (len < 2) { // 单条
            _this.execute_1(sqlUnits);
            return;
        }

        if (len < 3) {
            _this.execute_2(sqlUnits);
            return;
        }

        if (len < 4) {
            _this.execute_3(sqlUnits);
            return;
        }

        if (len < 5) {
            _this.execute_4(sqlUnits);
            return;
        }

        if (len < 6) {
            _this.execute_5(sqlUnits);
            return;
        }

        if (len < 11) {
            _this.execute_6_10(sqlUnits);
            return;
        }

        if (len < 20) {
            _this.execute_11_20(sqlUnits);
            return;
        }

        _this.execute_21_(sqlUnits);
        return;
    }

    @SpringGraph
    public void execute_1(List<SqlUnit> sqlUnits) {
        this.execute_in_thread(sqlUnits);
    }

    @SpringGraph
    public void execute_2(List<SqlUnit> sqlUnits) {
        this.execute_in_thread(sqlUnits);
    }

    @SpringGraph
    public void execute_3(List<SqlUnit> sqlUnits) {
        this.execute_in_thread(sqlUnits);
    }

    @SpringGraph
    public void execute_4(List<SqlUnit> sqlUnits) {
        this.execute_in_thread(sqlUnits);
    }

    @SpringGraph
    public void execute_5(List<SqlUnit> sqlUnits) {
        this.execute_in_thread(sqlUnits);
    }

    @SpringGraph
    public void execute_6_10(List<SqlUnit> sqlUnits) {
        this.execute_in_thread(sqlUnits);
    }

    @SpringGraph
    public void execute_11_20(List<SqlUnit> sqlUnits) {
        this.execute_in_thread(sqlUnits);
    }

    @SpringGraph
    public void execute_21_(List<SqlUnit> sqlUnits) {
        this.execute_in_thread(sqlUnits);
    }

    private void execute_in_thread(final List<SqlUnit> sqlUnits) {
        if (null == this.getJdbcExecuteService()) {
            throw new RuntimeException("#138 jdbcExecuteService is null, this=" + this);
        }

        final Pipeline pipeline = this.getPipeline();
        List<Callable<Object>> tasks = new ArrayList<Callable<Object>>();

        for (final SqlUnit sqlUnit : sqlUnits) {
            if (null == sqlUnit) {
                log.error("#146 sqlUnit=" + sqlUnit + ", this=" + this);
                continue;
            }

            if (SqlUnit.NULL.equals(sqlUnit)) {
                log.debug("#152 sqlUnit={}, this={}", sqlUnit, this);
                continue;
            }

            Runnable runnable = new Runnable() {
                public void run() {
                    JdbcDataConsumer.this.getJdbcExecuteService().executeUpdate(pipeline, sqlUnit.getSql(), sqlUnit.getParameters());

                    log.debug("#154 after executeUpdate, pipeline={}, sqlUnit={}, this={}", pipeline, sqlUnit, this);
                }
            };

            Callable<Object> callable = Executors.callable(runnable);
            tasks.add(callable);
        }

        // 全部执行完后返回
        ExecutorUtil.invokeAll(this.getExecutorService(), tasks);

        log.debug("#159 after invokeAll, executorService={}, tasks={}, this={}", this.getExecutorService(), tasks, this);
    }

    private JdbcExecuteService getJdbcExecuteService() {
        JdbcExecuteService jdbcExecuteService = SpringUtil.getBeanCached(JdbcExecuteService.class);

        if (null == jdbcExecuteService) {
            throw new RuntimeException("#175 jdbcExecuteService=" + jdbcExecuteService + ", this=" + this);
        }

        return jdbcExecuteService;
    }
}